//===----------------------------------------------------------------------===//
//
// This source file is part of the Soto for AWS open source project
//
// Copyright (c) 2017-2020 the Soto project authors
// Licensed under Apache License v2.0
//
// See LICENSE.txt for license information
// See CONTRIBUTORS.txt for the list of Soto project authors
//
// SPDX-License-Identifier: Apache-2.0
//
//===----------------------------------------------------------------------===//

import Atomics
import Logging
import NIOConcurrencyHelpers
import NIOCore
import NIOPosix
import SotoCore

extension S3ErrorType {
    public enum MultipartError: Error {
        /// CreateMultipartUpload did not return an uploadId
        case noUploadId
        /// Copy Source in multipartCopy has no content length
        case noCopyPartResult
        /// An empty download was returned
        case downloadEmpty(message: String)
        /// returned if an upload fails and `abortOnFail` is set to false. `uploadId` can be used in `resumeMultipartUpload`. `error` is the original error
        case abortedUpload(resumeRequest: S3.ResumeMultipartUploadRequest, error: Error)
    }
}

extension S3 {
    /// Resume Multipart upload request object. This is used as a paramter `resumeMultipartUpload`. It can only be generated by a `abortedUpload` error
    public struct ResumeMultipartUploadRequest: Sendable {
        let uploadRequest: CreateMultipartUploadRequest
        public let uploadId: String
        public let completedParts: [S3.CompletedPart]
    }

    /// error thrown during upload of parts
    struct MultipartUploadError: Swift.Error {
        let error: Swift.Error
        let completedParts: [S3.CompletedPart]
    }

    /// Multipart download of a file from S3.
    ///
    /// - parameters:
    ///     - input: The GetObjectRequest shape that contains the details of the object request.
    ///     - partSize: Size of each part to be downloaded
    ///     - concurrentDownloads: How many downloads can you have running at one time
    ///     - outputStream: Function to be called for each downloaded part. Called with data block and file size
    /// - returns: The complete file size once the multipart download has finished.
    public func multipartDownload(
        _ input: GetObjectRequest,
        partSize: Int = 5 * 1024 * 1024,
        concurrentDownloads: Int = 4,
        logger: Logger = AWSClient.loggingDisabled,
        outputStream: @escaping (ByteBuffer, Int64) async throws -> Void
    ) async throws -> Int64 {
        // get object size before downloading
        let headRequest = S3.HeadObjectRequest(
            bucket: input.bucket,
            ifMatch: input.ifMatch,
            ifModifiedSince: input.ifModifiedSince,
            ifNoneMatch: input.ifNoneMatch,
            ifUnmodifiedSince: input.ifUnmodifiedSince,
            key: input.key,
            requestPayer: input.requestPayer,
            sseCustomerAlgorithm: input.sseCustomerAlgorithm,
            sseCustomerKey: input.sseCustomerKey,
            sseCustomerKeyMD5: input.sseCustomerKeyMD5,
            versionId: input.versionId
        )
        let object = try await headObject(headRequest, logger: logger)
        guard let contentLength = object.contentLength, contentLength > 0 else {
            throw S3ErrorType.MultipartError.downloadEmpty(message: "Content length is unexpectedly zero")
        }

        try await withThrowingTaskGroup(of: (Int, ByteBuffer).self) { group in
            /// Structure used to store downloaded buffers and then save them as and when
            /// needed
            struct DownloadedBuffers {
                let outputStream: (ByteBuffer) async throws -> Void
                var buffers: [ByteBuffer?]
                var bufferSavedIndex: Int

                init(numberOfBuffers: Int, outputStream: @escaping (ByteBuffer) async throws -> Void) {
                    self.outputStream = outputStream
                    self.buffers = Array(repeating: nil, count: numberOfBuffers)
                    self.bufferSavedIndex = 0
                }

                mutating func saveBuffer(index: Int, buffer: ByteBuffer) async throws {
                    assert(index >= 0 && index < self.buffers.count)
                    self.buffers[index] = buffer
                    while self.bufferSavedIndex < self.buffers.count, let bufferToSave = self.buffers[bufferSavedIndex] {
                        self.buffers[self.bufferSavedIndex] = nil
                        self.bufferSavedIndex += 1
                        try await self.outputStream(bufferToSave)
                    }
                }
            }
            let partSize64: Int64 = numericCast(partSize)
            var count = 0
            var offset: Int64 = 0
            let numberOfParts: Int = numericCast((contentLength - 1) / partSize64) + 1
            var downloadBuffers = DownloadedBuffers(numberOfBuffers: numberOfParts) { buffer in
                try await outputStream(buffer, contentLength)
            }
            // while we still have parts to download
            while count < numberOfParts {
                if count > concurrentDownloads {
                    // if count is greater than concurrentDownloads then start waiting for
                    // parts that have downloaded to save them
                    if let (index, buffer) = try await group.next() {
                        // save the buffer
                        try await downloadBuffers.saveBuffer(index: index, buffer: buffer)
                    }
                }
                let index = count
                let currentPartSize = min(partSize64, contentLength - offset)
                let currentOffset = offset
                // add task downloading from S3
                group.addTask {
                    let range = "bytes=\(currentOffset)-\(currentOffset + currentPartSize - 1)"
                    let getRequest = S3.GetObjectRequest(
                        bucket: input.bucket,
                        key: input.key,
                        range: range,
                        sseCustomerAlgorithm: input.sseCustomerAlgorithm,
                        sseCustomerKey: input.sseCustomerKey,
                        sseCustomerKeyMD5: input.sseCustomerKeyMD5,
                        versionId: input.versionId
                    )
                    let getObjectOutput = try await getObject(getRequest, logger: logger)
                    let buffer = try await getObjectOutput.body.collect(upTo: partSize)
                    return (index, buffer)
                }
                offset += partSize64
                count += 1
            }

            // save the remaining parts
            for try await (index, buffer) in group {
                // save the buffer
                try await downloadBuffers.saveBuffer(index: index, buffer: buffer)
            }
        }
        return contentLength
    }

    /// Multipart download of a file from S3.
    ///
    /// - parameters:
    ///     - input: The GetObjectRequest shape that contains the details of the object request.
    ///     - partSize: Size of each part to be downloaded
    ///     - filename: Filename to save download to
    ///     - concurrentDownloads: How many downloads can you have running at one time
    ///     - threadPool: Thread pool used to save file
    ///     - logger: logger
    ///     - progress: Callback that returns the progress of the download. It is called after each part is downloaded with a value
    ///         between 0.0 and 1.0 indicating how far the download is complete (1.0 meaning finished).
    /// - returns: The complete file size once the multipart download has finished.
    public func multipartDownload(
        _ input: GetObjectRequest,
        partSize: Int = 5 * 1024 * 1024,
        filename: String,
        concurrentDownloads: Int = 4,
        threadPool: NIOThreadPool = .singleton,
        logger: Logger = AWSClient.loggingDisabled,
        progress: @escaping @Sendable (Double) async throws -> Void = { _ in }
    ) async throws -> Int64 {
        let fileIO = NonBlockingFileIO(threadPool: threadPool)
        return try await fileIO.withFileHandle(path: filename, mode: .write, flags: .allowFileCreation()) { fileHandle in
            let progressValue = ManagedAtomic(0)

            let result = try await self.multipartDownload(
                input,
                partSize: partSize,
                concurrentDownloads: concurrentDownloads,
                logger: logger
            ) { byteBuffer, fileSize in
                let bufferSize = byteBuffer.readableBytes
                try await fileIO.write(fileHandle: fileHandle, buffer: byteBuffer)
                let progressIntValue = progressValue.wrappingIncrementThenLoad(by: bufferSize, ordering: .relaxed)
                try await progress(Double(progressIntValue) / Double(fileSize))
            }
            return result
        }
    }

    /// Multipart upload of ByteBuffer to S3.
    ///
    /// Uploads buffer using multipart upload commands. If you want the function to not abort the multipart upload when it
    /// receives an error then set `abortOnFail` to false. With this you can then use `resumeMultipartUpload` to resume
    /// the failed upload. If you set `abortOnFail` to false but don't call `resumeMultipartUpload` on failure you will have
    /// to call `abortMultipartUpload` yourself.
    ///
    /// - parameters:
    ///     - input: The CreateMultipartUploadRequest structure that contains the details about the upload
    ///     - partSize: Size of each part to upload. This has to be at least 5MB
    ///     - filename: Full path of file to upload
    ///     - concurrentUploads: Number of uploads to run at one time
    ///     - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after
    ///         a fail this should be set to false
    ///     - logger: logger
    ///     - progress: Callback that returns the progress of the upload. It is called after each part and is called with how
    ///         many bytes have been uploaded so far.
    /// - returns: Output from CompleteMultipartUpload.
    public func multipartUpload(
        _ input: CreateMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        buffer: ByteBuffer,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        logger: Logger = AWSClient.loggingDisabled,
        progress: (@Sendable (Int) async throws -> Void)? = nil
    ) async throws -> CompleteMultipartUploadOutput {
        try await self.multipartUpload(
            input,
            partSize: partSize,
            bufferSequence: buffer.asyncSequence(chunkSize: partSize),
            concurrentUploads: concurrentUploads,
            abortOnFail: abortOnFail,
            logger: logger,
            progress: progress
        )
    }

    /// Multipart upload of file to S3.
    ///
    /// Uploads file using multipart upload commands. If you want the function to not abort the multipart upload when it receives
    /// an error then set `abortOnFail` to false. With this you can then use `resumeMultipartUpload` to resume the failed upload.
    /// If you set `abortOnFail` to false but don't call `resumeMultipartUpload` on failure you will have to call `abortMultipartUpload`
    /// yourself.
    ///
    /// - parameters:
    ///     - input: The CreateMultipartUploadRequest structure that contains the details about the upload
    ///     - partSize: Size of each part to upload. This has to be at least 5MB
    ///     - filename: Full path of file to upload
    ///     - concurrentUploads: Number of uploads to run at one time
    ///     - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after a fail this should
    ///         be set to false
    ///     - threadPool: Thread pool used to load file
    ///     - logger: logger
    ///     - progress: Callback that returns the progress of the upload. It is called after each part is uploaded with a value between
    ///         0.0 and 1.0 indicating how far the upload is complete (1.0 meaning finished).
    /// - returns: The output from CompleteMultipartUploadOutput once the multipart upload has finished.
    public func multipartUpload(
        _ input: CreateMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        filename: String,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        threadPool: NIOThreadPool = .singleton,
        logger: Logger = AWSClient.loggingDisabled,
        progress: @escaping @Sendable (Double) async throws -> Void = { _ in }
    ) async throws -> CompleteMultipartUploadOutput {
        let fileIO = NonBlockingFileIO(threadPool: threadPool)
        return try await fileIO.withFileRegion(path: filename) { fileRegion in
            let fileSequence = FileByteBufferAsyncSequence(
                fileRegion.fileHandle,
                fileIO: fileIO,
                chunkSize: partSize
            )
            let length = Double(fileRegion.readableBytes)
            @Sendable func percentProgress(_ value: Int) async throws {
                try await progress(Double(value) / length)
            }
            return try await self.multipartUpload(
                input,
                partSize: partSize,
                bufferSequence: fileSequence,
                concurrentUploads: concurrentUploads,
                abortOnFail: abortOnFail,
                logger: logger,
                progress: percentProgress
            )
        }
    }

    ///  Resume upload of failed multipart upload
    ///
    /// - Parameters:
    ///   - input: The CreateMultipartUploadRequest structure that contains the details about the upload
    ///   - partSize: Size of each part to upload. Should be the same as the original upload
    ///   - bufferSequence: Sequence of ByteBuffers to upload
    ///   - concurrentUploads: Number of uploads to run at one time
    ///   - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after
    ///         a fail this should be set to false
    ///   - logger: logger
    ///   - progress: Callback that returns the progress of the upload. It is called after each part and is called with how
    ///         many bytes have been uploaded so far.
    /// - Returns: Output from CompleteMultipartUpload.
    public func resumeMultipartUpload(
        _ input: ResumeMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        buffer: ByteBuffer,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        logger: Logger = AWSClient.loggingDisabled,
        progress: (@Sendable (Int) async throws -> Void)? = nil
    ) async throws -> CompleteMultipartUploadOutput {
        try await self.resumeMultipartUpload(
            input,
            partSize: partSize,
            bufferSequence: buffer.asyncSequence(chunkSize: partSize),
            concurrentUploads: concurrentUploads,
            abortOnFail: abortOnFail,
            logger: logger,
            progress: progress
        )
    }

    /// Resume multipart upload of file to S3.
    ///
    /// Call this with `ResumeMultipartUploadRequest`returned by the failed multipart upload. Make sure you are using the same
    /// `partSize`and `filename` as you used when calling `multipartUpload`. `
    ///
    /// - parameters:
    ///     - input: The `ResumeMultipartUploadRequest` structure returned in upload fail error from previous upload call
    ///     - partSize: Size of each part to upload. This has to be at least 5MB
    ///     - filename: Full path of file to upload
    ///     - concurrentUploads: Number of uploads to run at one time
    ///     - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after a fail
    ///         this should be set to false
    ///     - threadPool: Thread pool used to load file
    ///     - progress: Callback that returns the progress of the upload. It is called after each part is uploaded with a value
    ///         between 0.0 and 1.0 indicating how far the upload is complete (1.0 meaning finished).
    /// - returns: Output from CompleteMultipartUpload.
    public func resumeMultipartUpload(
        _ input: ResumeMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        filename: String,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        threadPool: NIOThreadPool = .singleton,
        logger: Logger = AWSClient.loggingDisabled,
        progress: @escaping @Sendable (Double) async throws -> Void = { _ in }
    ) async throws -> CompleteMultipartUploadOutput {
        let fileIO = NonBlockingFileIO(threadPool: threadPool)
        return try await fileIO.withFileRegion(path: filename) { fileRegion in
            let fileSequence = FileByteBufferAsyncSequence(
                fileRegion.fileHandle,
                fileIO: fileIO,
                chunkSize: partSize
            )
            // let chunks = fileHandle.readChunks(in: ..., chunkLength: .bytes(numericCast(partSize)))
            let length = Double(fileRegion.readableBytes)
            @Sendable func percentProgress(_ value: Int) async throws {
                try await progress(Double(value) / length)
            }
            return try await self.resumeMultipartUpload(
                input,
                partSize: partSize,
                bufferSequence: fileSequence,
                concurrentUploads: concurrentUploads,
                abortOnFail: abortOnFail,
                logger: logger,
                progress: percentProgress
            )
        }
    }

    /// Multipart copy of file to S3. Currently this only works within one region as it uses HeadObject to read the source file size
    ///
    /// - parameters:
    ///     - input: The CopyObjectRequest structure that contains the details about the copy
    ///     - partSize: Size of each part to copy. This has to be at least 5MB
    ///     - logger: logger
    /// - returns: The output from the CompleteMultipartUploadOutput once the multipart upload has finished.
    public func multipartCopy(
        _ input: CopyObjectRequest,
        partSize: Int = 8 * 1024 * 1024,
        logger: Logger = AWSClient.loggingDisabled
    ) async throws -> CompleteMultipartUploadOutput {
        // get object bucket, key and version from copySource
        guard let copySourceValues = getBucketKeyVersion(from: input.copySource) else { throw AWSClientError.validationError }

        // get object size from headObject
        let head = try await self.headObject(
            .init(bucket: copySourceValues.bucket, key: copySourceValues.key, versionId: copySourceValues.versionId),
            logger: logger
        )
        let objectSize = head.contentLength ?? 0

        // initialize multipart upload
        let request: CreateMultipartUploadRequest = .init(
            acl: input.acl,
            bucket: input.bucket,
            cacheControl: input.cacheControl,
            contentDisposition: input.contentDisposition,
            contentEncoding: input.contentEncoding,
            contentLanguage: input.contentLanguage,
            contentType: input.contentType,
            expectedBucketOwner: input.expectedBucketOwner,
            expires: input.expires,
            grantFullControl: input.grantFullControl,
            grantRead: input.grantRead,
            grantReadACP: input.grantReadACP,
            grantWriteACP: input.grantWriteACP,
            key: input.key,
            metadata: input.metadata,
            objectLockLegalHoldStatus: input.objectLockLegalHoldStatus,
            objectLockMode: input.objectLockMode,
            objectLockRetainUntilDate: input.objectLockRetainUntilDate,
            requestPayer: input.requestPayer,
            serverSideEncryption: input.serverSideEncryption,
            sseCustomerAlgorithm: input.sseCustomerAlgorithm,
            sseCustomerKey: input.sseCustomerKey,
            sseCustomerKeyMD5: input.sseCustomerKeyMD5,
            ssekmsEncryptionContext: input.ssekmsEncryptionContext,
            ssekmsKeyId: input.ssekmsKeyId,
            storageClass: input.storageClass,
            tagging: input.tagging,
            websiteRedirectLocation: input.websiteRedirectLocation
        )
        let uploadResponse = try await createMultipartUpload(request, logger: logger)
        guard let uploadId = uploadResponse.uploadId else {
            throw S3ErrorType.MultipartError.noUploadId
        }

        do {
            // calculate number of upload part calls and the size of the final upload
            let numParts = ((Int(objectSize) - 1) / partSize) + 1
            let finalPartSize = Int(objectSize) - (numParts - 1) * partSize

            // create array of upload part requests.
            let uploadPartRequests: [UploadPartCopyRequest] = (1...numParts).map { part in
                let copyRange =
                    if part != numParts {
                        "bytes=\((part - 1) * partSize)-\(part * partSize - 1)"
                    } else {
                        "bytes=\((part - 1) * partSize)-\((part - 1) * partSize + finalPartSize - 1)"
                    }
                return .init(
                    bucket: input.bucket,
                    copySource: input.copySource,
                    copySourceRange: copyRange,
                    copySourceSSECustomerAlgorithm: input.copySourceSSECustomerAlgorithm,
                    copySourceSSECustomerKey: input.copySourceSSECustomerKey,
                    copySourceSSECustomerKeyMD5: input.copySourceSSECustomerKeyMD5,
                    expectedBucketOwner: input.expectedBucketOwner,
                    expectedSourceBucketOwner: input.expectedSourceBucketOwner,
                    key: input.key,
                    partNumber: part,
                    requestPayer: input.requestPayer,
                    sseCustomerAlgorithm: input.sseCustomerAlgorithm,
                    sseCustomerKey: input.sseCustomerKey,
                    sseCustomerKeyMD5: input.sseCustomerKeyMD5,
                    uploadId: uploadId
                )
            }
            // send upload part copy requests to AWS
            let parts: [S3.CompletedPart] = try await uploadPartRequests.concurrentMap(maxConcurrentTasks: 8) {
                let response = try await self.uploadPartCopy($0, logger: logger)
                return S3.CompletedPart(eTag: response.copyPartResult.eTag, partNumber: $0.partNumber)
            }
            // complete upload
            let completeRequest = S3.CompleteMultipartUploadRequest(
                bucket: input.bucket,
                key: input.key,
                multipartUpload: S3.CompletedMultipartUpload(parts: parts),
                requestPayer: input.requestPayer,
                uploadId: uploadId
            )
            return try await self.completeMultipartUpload(completeRequest, logger: logger)
        } catch {
            // if failure then abort the multipart upload
            let request = S3.AbortMultipartUploadRequest(
                bucket: input.bucket,
                key: input.key,
                requestPayer: input.requestPayer,
                uploadId: uploadId
            )
            _ = try await self.abortMultipartUpload(request, logger: logger)
            throw error
        }
    }
}

/// AsyncSequence version of multipart upload
extension S3 {
    /// Multipart upload of AsyncSequence to S3.
    ///
    /// Uploads file using multipart upload commands. If you want the function to not abort the multipart upload when it
    /// receives an error then set `abortOnFail` to false. With this you can then use `resumeMultipartUpload` to resume
    /// the failed upload. If you set `abortOnFail` to false but don't call `resumeMultipartUpload` on failure you will have
    /// to call `abortMultipartUpload` yourself.
    ///
    /// - parameters:
    ///     - input: The CreateMultipartUploadRequest structure that contains the details about the upload
    ///     - partSize: Size of each part to upload. This has to be at least 5MB
    ///     - filename: Full path of file to upload
    ///     - concurrentUploads: Number of uploads to run at one time
    ///     - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after
    ///         a fail this should be set to false
    ///     - logger: logger
    ///     - progress: Callback that returns the progress of the upload. It is called after each part and is called with how
    ///         many bytes have been uploaded so far.
    /// - returns: Output from CompleteMultipartUpload.
    public func multipartUpload<ByteBufferSequence: AsyncSequence>(
        _ input: CreateMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        bufferSequence: ByteBufferSequence,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        logger: Logger = AWSClient.loggingDisabled,
        progress: (@Sendable (Int) async throws -> Void)? = nil
    ) async throws -> CompleteMultipartUploadOutput where ByteBufferSequence.Element == ByteBuffer {
        // initialize multipart upload
        let upload = try await createMultipartUpload(input, logger: logger)
        guard let uploadId = upload.uploadId else {
            throw S3ErrorType.MultipartError.noUploadId
        }

        do {
            // upload all the parts
            let parts = try await self.multipartUploadParts(
                input,
                uploadId: uploadId,
                partSequence: bufferSequence.fixedSizeSequence(chunkSize: partSize).enumerated(),
                concurrentUploads: concurrentUploads,
                initialProgress: 0,
                logger: logger,
                progress: progress
            )

            // complete multipart upload
            let request = S3.CompleteMultipartUploadRequest(
                bucket: input.bucket,
                key: input.key,
                multipartUpload: S3.CompletedMultipartUpload(parts: parts),
                requestPayer: input.requestPayer,
                uploadId: uploadId
            )
            do {
                return try await self.completeMultipartUpload(request, logger: logger)
            } catch {
                throw MultipartUploadError(error: error, completedParts: parts)
            }
        } catch {
            guard abortOnFail else {
                // if error is MultipartUploadError then we have completed uploading some parts and should include that in the error
                if let error = error as? MultipartUploadError {
                    throw S3ErrorType.MultipartError.abortedUpload(
                        resumeRequest: .init(uploadRequest: input, uploadId: uploadId, completedParts: error.completedParts),
                        error: error.error
                    )
                } else {
                    throw S3ErrorType.MultipartError.abortedUpload(
                        resumeRequest: .init(uploadRequest: input, uploadId: uploadId, completedParts: []),
                        error: error
                    )
                }
            }
            // if failure then abort the multipart upload
            let request = S3.AbortMultipartUploadRequest(
                bucket: input.bucket,
                key: input.key,
                requestPayer: input.requestPayer,
                uploadId: uploadId
            )
            _ = try await self.abortMultipartUpload(request, logger: logger)
            throw error
        }
    }

    ///  Resume upload of failed multipart upload
    ///
    /// - Parameters:
    ///   - input: The CreateMultipartUploadRequest structure that contains the details about the upload
    ///   - partSize: Size of each part to upload. Should be the same as the original upload
    ///   - bufferSequence: Sequence of ByteBuffers to upload
    ///   - concurrentUploads: Number of uploads to run at one time
    ///   - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after
    ///         a fail this should be set to false
    ///   - logger: logger
    ///   - progress: Callback that returns the progress of the upload. It is called after each part and is called with how
    ///         many bytes have been uploaded so far.
    /// - Returns: Output from CompleteMultipartUpload.
    public func resumeMultipartUpload<ByteBufferSequence: AsyncSequence>(
        _ input: ResumeMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        bufferSequence: ByteBufferSequence,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        logger: Logger = AWSClient.loggingDisabled,
        progress: (@Sendable (Int) async throws -> Void)? = nil
    ) async throws -> CompleteMultipartUploadOutput where ByteBufferSequence.Element == ByteBuffer {
        // upload all the parts
        let partsSet = Set<Int>(input.completedParts.map { $0.partNumber! - 1 })
        let partSequence =
            bufferSequence
            .fixedSizeSequence(chunkSize: partSize)
            .enumerated()
            .filter { !partsSet.contains($0.0) }

        return try await self.resumeMultipartUpload(
            input,
            partSize: partSize,
            partSequence: partSequence,
            concurrentUploads: concurrentUploads,
            abortOnFail: abortOnFail,
            logger: logger,
            progress: progress
        )
    }

    ///  Resume upload of failed multipart upload
    ///
    /// - Parameters:
    ///   - input: The CreateMultipartUploadRequest structure that contains the details about the upload
    ///   - partSize: Size of each part to upload. Should be the same as the original upload
    ///   - bufferSequence: Sequence of ByteBuffers to upload
    ///   - concurrentUploads: Number of uploads to run at one time
    ///   - abortOnFail: Whether should abort multipart upload if it fails. If you want to attempt to resume after
    ///         a fail this should be set to false
    ///   - logger: logger
    ///   - progress: Callback that returns the progress of the upload. It is called after each part and is called with how
    ///         many bytes have been uploaded so far.
    /// - Returns: Output from CompleteMultipartUpload.
    public func resumeMultipartUpload<PartsSequence: AsyncSequence>(
        _ input: ResumeMultipartUploadRequest,
        partSize: Int = 5 * 1024 * 1024,
        partSequence: PartsSequence,
        concurrentUploads: Int = 4,
        abortOnFail: Bool = true,
        logger: Logger = AWSClient.loggingDisabled,
        progress: (@Sendable (Int) async throws -> Void)? = nil
    ) async throws -> CompleteMultipartUploadOutput where PartsSequence.Element == (Int, ByteBuffer) {
        let uploadRequest = input.uploadRequest

        do {
            // upload all the parts
            let parts = try await self.multipartUploadParts(
                uploadRequest,
                uploadId: input.uploadId,
                partSequence: partSequence,
                concurrentUploads: concurrentUploads,
                initialProgress: input.completedParts.count * partSize,
                logger: logger,
                progress: progress
            )
            // combine array of already uploaded parts prior to the resume with the parts just uploaded
            let completedParts = (input.completedParts + parts).sorted { $0.partNumber! < $1.partNumber! }

            let request = S3.CompleteMultipartUploadRequest(
                bucket: uploadRequest.bucket,
                key: uploadRequest.key,
                multipartUpload: S3.CompletedMultipartUpload(parts: completedParts),
                requestPayer: uploadRequest.requestPayer,
                uploadId: input.uploadId
            )
            do {
                return try await self.completeMultipartUpload(request, logger: logger)
            } catch {
                throw MultipartUploadError(error: error, completedParts: completedParts)
            }
        } catch {
            guard abortOnFail else {
                // if error is MultipartUploadError then we have completed uploading some parts and should include that in the error
                if let error = error as? MultipartUploadError {
                    throw S3ErrorType.MultipartError.abortedUpload(
                        resumeRequest: .init(uploadRequest: uploadRequest, uploadId: input.uploadId, completedParts: error.completedParts),
                        error: error.error
                    )
                } else {
                    throw S3ErrorType.MultipartError.abortedUpload(
                        resumeRequest: .init(uploadRequest: uploadRequest, uploadId: input.uploadId, completedParts: []),
                        error: error
                    )
                }
            }
            // if failure then abort the multipart upload
            let request = S3.AbortMultipartUploadRequest(
                bucket: uploadRequest.bucket,
                key: uploadRequest.key,
                requestPayer: uploadRequest.requestPayer,
                uploadId: input.uploadId
            )
            _ = try await self.abortMultipartUpload(request, logger: logger)
            throw error
        }
    }
}

// Internal functions used by multipart upload
extension S3 {
    /// Used internally in multipartUpload, loads all the parts once the multipart upload has been initiated
    ///
    /// - Parameters:
    ///   - input: multipart upload request
    ///   - uploadId: upload id
    ///   - partSequence: AsyncSequence supplying fixed size ByteBuffers
    ///   - concurrentUploads: Number of uploads to run at one time
    ///   - logger: logger
    ///   - progress: Progress function updated with accumulated amount uploaded.
    /// - Returns: Array of completed parts
    func multipartUploadParts<PartSequence: AsyncSequence>(
        _ input: CreateMultipartUploadRequest,
        uploadId: String,
        partSequence: PartSequence,
        concurrentUploads: Int,
        initialProgress: Int,
        logger: Logger,
        progress: (@Sendable (Int) async throws -> Void)?
    ) async throws -> [S3.CompletedPart] where PartSequence.Element == (Int, ByteBuffer) {
        var newProgress: (@Sendable (Int) async throws -> Void)?
        if let progress {
            let size = ManagedAtomic(initialProgress)
            @Sendable func accumulatingProgress(_ amount: Int) async throws {
                let totalSize = size.wrappingIncrementThenLoad(by: amount, ordering: .relaxed)
                try await progress(totalSize)
            }
            newProgress = accumulatingProgress
        }

        return try await withThrowingTaskGroup(of: (Int, S3.CompletedPart).self) { group in
            var results = ContiguousArray<(Int, S3.CompletedPart)>()

            var count = 0
            for try await (index, buffer) in partSequence {
                count += 1
                // once we have kicked off `concurrentUploads` tasks we can start waiting for a task to finish before
                // starting another
                if count > concurrentUploads {
                    if let element = try await group.next() {
                        results.append(element)
                    }
                }
                let body: AWSHTTPBody =
                    if let progress = newProgress {
                        .init(
                            asyncSequence: buffer.asyncSequence(chunkSize: 64 * 1024).reportProgress(reportFn: progress),
                            length: buffer.readableBytes
                        )
                    } else {
                        .init(asyncSequence: buffer.asyncSequence(chunkSize: 64 * 1024), length: buffer.readableBytes)
                    }
                group.addTask {
                    // Multipart uploads part numbers start at 1 not 0
                    let request = S3.UploadPartRequest(
                        body: body,
                        bucket: input.bucket,
                        key: input.key,
                        partNumber: index + 1,
                        requestPayer: input.requestPayer,
                        sseCustomerAlgorithm: input.sseCustomerAlgorithm,
                        sseCustomerKey: input.sseCustomerKey,
                        sseCustomerKeyMD5: input.sseCustomerKeyMD5,
                        uploadId: uploadId
                    )
                    let uploadOutput = try await self.uploadPart(request, logger: logger)
                    let part = S3.CompletedPart(eTag: uploadOutput.eTag, partNumber: index + 1)

                    return (index, part)
                }
            }
            // if no parts were uploaded and this is not called from resumeMultipartUpload then
            // upload an empty part
            if count == 0, initialProgress == 0 {
                group.addTask {
                    let request = S3.UploadPartRequest(
                        body: .init(),
                        bucket: input.bucket,
                        key: input.key,
                        partNumber: 1,
                        requestPayer: input.requestPayer,
                        sseCustomerAlgorithm: input.sseCustomerAlgorithm,
                        sseCustomerKey: input.sseCustomerKey,
                        sseCustomerKeyMD5: input.sseCustomerKeyMD5,
                        uploadId: uploadId
                    )
                    let uploadOutput = try await self.uploadPart(request, logger: logger)
                    let part = S3.CompletedPart(eTag: uploadOutput.eTag, partNumber: 1)

                    return (0, part)
                }
            }
            do {
                while let element = try await group.next() {
                    results.append(element)
                }
            } catch {
                throw MultipartUploadError(error: error, completedParts: results.sorted { $0.0 < $1.0 }.map(\.1))
            }
            // construct final array and fill in elements
            return results.sorted { $0.0 < $1.0 }.map(\.1)
        }
    }

    // from bucket, key and version id from a copySource string
    func getBucketKeyVersion(from copySource: String) -> (bucket: String, key: String, versionId: String?)? {
        // drop first slash if it exists
        let path =
            if copySource.first == "/" {
                copySource.dropFirst()
            } else {
                Substring(copySource)
            }
        // find first slash
        guard let slashIndex = path.firstIndex(of: "/") else { return nil }
        // find first question mark
        let questionMarkIndex = path.firstIndex(of: "?")
        // key is between first slash and either end index or question mark index
        let keyStartIndex = path.index(after: slashIndex)
        let keyEndIndex = questionMarkIndex ?? path.endIndex
        // bucket is from start to first slash
        let bucket = path[path.startIndex..<slashIndex]
        let key = path[keyStartIndex..<keyEndIndex]
        var versionId: String?
        // if there was a question mark then check for version id
        if questionMarkIndex != nil {
            let versionParameter = path[questionMarkIndex!..<path.endIndex]
            if versionParameter.hasPrefix("?versionId=") {
                versionId = String(versionParameter.dropFirst(11))
            } else {
                return nil
            }
        }
        return (bucket: String(bucket), key: String(key), versionId: versionId)
    }
}
